package com.example.netty;


import com.alibaba.fastjson.JSONObject;
import com.example.service.impl.NettySendSysNotice;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/*
      连接用户管理

 */

@Slf4j
public class nettyUserManager {


    private static ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock(true);

    //对象和通道对应。
    public static ConcurrentMap<Channel, nettyUserInfo> userInfos = new ConcurrentHashMap<>();

    private static AtomicInteger userCount = new AtomicInteger(0);

    public static void addChannel(Channel channel) {
        String remoteAddr = NettyUtil.parseChannelRemoteAddr(channel);
        if (!channel.isActive()) {
            log.error("channel is not active, address: {}", remoteAddr);
        }
        nettyUserInfo nettyUserInfo = new nettyUserInfo();
        nettyUserInfo.setAddr(remoteAddr);
        nettyUserInfo.setChannel(channel);
        nettyUserInfo.setTime(System.currentTimeMillis());
        userInfos.put(channel, nettyUserInfo);
    }

    public static boolean saveUser(Channel channel, JSONObject jsonObj) {
        String nick = jsonObj.getString("nick");
        String userid = jsonObj.getString("userId");
        nettyUserInfo userInfo = userInfos.get(channel);
        if (userInfo == null) {
            return false;
        }
        if (!channel.isActive()) {
            log.error("channel is not active, address: {}, nick: {}", userInfo.getAddr(), nick);
            return false;
        }
        // 增加一个认证用户
        userCount.incrementAndGet();
        userInfo.setNick(nick);
        userInfo.setUserId(userid);
        userInfo.setAuth(true);
        userInfo.setTime(System.currentTimeMillis());
        return true;
    }

    public static void broadCastPing() {
        try {
            rwLock.readLock().lock();
            log.info("broadCastPing userCount: {}", userCount.intValue());
            Set<Channel> keySet = userInfos.keySet();
            for (Channel ch : keySet) {
                nettyUserInfo userInfo = userInfos.get(ch);
                if (userInfo == null || !userInfo.isAuth()) {
                    continue;
                }
                ch.writeAndFlush(new TextWebSocketFrame(ChatProto.buildPingProto()));
            }
        } finally {
            rwLock.readLock().unlock();
        }
    }


    /**
     * 发送系统消息
     *
     * @param code
     * @param mess
     */
    public static void sendInfo(Channel channel, int code, Object mess) {
        channel.writeAndFlush(new TextWebSocketFrame(ChatProto.buildSystProto(code, mess)));
    }
    /**
     * 广播系统消息
     */
    public static void broadCastInfo(int code, Object mess) {
        try {
            rwLock.readLock().lock();
            Set<Channel> keySet = userInfos.keySet();
            for (Channel ch : keySet) {
                nettyUserInfo userInfo = userInfos.get(ch);
                log.info("客户端登陆，推送消息"+userInfo.toString());
                if (userInfo == null || !userInfo.isAuth()) {
                    continue;
                }
                if (userInfo.getUserId()!=null){
                    new NettySendSysNotice().sysNotionMsg(userInfo);
                }
                ch.writeAndFlush(new TextWebSocketFrame(ChatProto.buildSystProto(code, mess)));
            }
        } finally {
            rwLock.readLock().unlock();
        }
    }

    public static int getAuthUserCount() {
        return userCount.get();
    }

    /**
     * 从缓存中移除Channel，并且关闭Channel
     *
     * @param channel
     */
    public static void removeChannel(Channel channel) {
        try {
            //移出频道
            log.warn("channel will be remove, address is :{}", NettyUtil.parseChannelRemoteAddr(channel));
            rwLock.writeLock().lock();
            channel.close();
            nettyUserInfo userInfo = userInfos.get(channel);
            if (userInfo != null) {
                //从集合中删除该频道的用户信息。
                nettyUserInfo tmp = userInfos.remove(channel);
                if (tmp != null && tmp.isAuth()) {
                    // 减去一个认证用户
                    userCount.decrementAndGet();
                }
            }
        } finally {
            rwLock.writeLock().unlock();
        }

    }
    public static nettyUserInfo getUserInfo(Channel channel) {
        return userInfos.get(channel);
    }


    /**
     * 广播普通消息
     *
     * @param message
     */
    /* // UID // 昵称*/
    public static void broadcastMess(String uid, String nick, String message) {
        if (StringUtils.isNotEmpty(message)) {
            try {
                /*消息不为空，加上读锁*/
                rwLock.readLock().lock();
                /*ConcurrentMap<Channel, UserInfo> userInfos*/
                /*拿到所有注册通道KEY值*/
                Set<Channel> keySet = userInfos.keySet();
                /*遍历通道KRY拿到所有连接通道用户*/
                for (Channel ch : keySet) {
//                    拿到连接用户信息。
                    nettyUserInfo userInfo = userInfos.get(ch);
                    //判断用户为空，或者没有认证的。继续循环
                    if (userInfo == null || !userInfo.isAuth()) {
                        continue;
                    }
                    ch.writeAndFlush(new TextWebSocketFrame(ChatProto.buildMessProto(uid, nick, message)));
                }
            } finally {
                rwLock.readLock().unlock();
            }
        }
    }


}
